package com.wyl.rpc.server;

import com.wyl.rpc.codec.RpcRequest;
import com.wyl.rpc.codec.RpcResponse;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import net.sf.cglib.reflect.FastClass;
import net.sf.cglib.reflect.FastMethod;

import java.lang.reflect.InvocationTargetException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: wyl
 * @date: 2021/8/17
 * @des:
 */
@Slf4j
public class RpcSeverHandler extends SimpleChannelInboundHandler<RpcRequest> {

    /**
     * key: 接口名称
     * value: 具体实现类的实例
     */
    private Map<String, Object> handlerMap;

    /**
     * 定义线程池，用于异步执行
     */
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(16, 16, 600L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(65536));

    public RpcSeverHandler(Map<String, Object> handlerMap) {
        this.handlerMap = handlerMap;
    }

    /**
     * 解析rpcRequest，返回执行结果给客户端
     *
     * @param ctx
     * @param rpcRequest
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest rpcRequest) throws Exception {
        /*
        1. 解析rpcRequest
        2. 从handlerMap中找到具体的接口(key)所绑定的实例(bean)
        3. 通过反射cglib调用具体方法，传递相关参数执行逻辑
        4. 返回响应信息给调用方
         */
        executor.submit(new Runnable() {
            @Override
            public void run() {
                RpcResponse response = new RpcResponse();
                response.setRequestId(rpcRequest.getRequestId());
                try {
                    Object result = handle(rpcRequest);
                    response.setResult(result);
                } catch (Throwable t) {
                    response.setThrowable(t);
                    log.error("rpc service handle request Throwable: " + t);
                }

                ctx.writeAndFlush(response).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (future.isSuccess()) {
                            // afterRpcHook 可以添加后置处理方法
                        }
                    }
                });

            }
        });
    }

    /**
     * 解析request请求并且去通过反射获取具体的本地服务实例，然后执行具体的方法
     *
     * @param request
     * @return
     * @throws InvocationTargetException
     */
    private Object handle(RpcRequest request) throws InvocationTargetException {
        String className = request.getClassName();
        Object serviceRef = handlerMap.get(className);
        Class<?> serviceClass = serviceRef.getClass();
        String methodName = request.getMethodName();
        Class<?>[] paramterTypes = request.getParamterTypes();
        Object[] paramters = request.getParamters();

        // JDK relect

        // cglib
        FastClass serviceFastClass = FastClass.create(serviceClass);
        FastMethod servicFastMethod = serviceFastClass.getMethod(methodName, paramterTypes);
        return servicFastMethod.invoke(serviceRef, paramters);
    }

    /**
     * 异常处理关闭连接
     *
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("server caught Throwable: " + cause);
        ctx.close();
    }
}
